package net.sf.calibur.transport.mina;

import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import net.sf.calibur.msgbuf.Command;
import net.sf.calibur.msgbuf.MessageBean;
import net.sf.calibur.transport.TransportException;
import net.sf.calibur.transport.Session;
import net.sf.calibur.transport.SessionManager;
import net.sf.calibur.transport.SessionState;
import net.sf.calibur.transport.message.CommandMessage;
import net.sf.calibur.transport.message.Message;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MinaSessionImpl implements Session {
	private final static AtomicInteger ID = new AtomicInteger(0);

	private MinaTcpTransporter parent;
	private IoSession ioSession;
	private SessionManager sessionManager;
	private String key;
	private boolean crypto;
	private boolean closed;
	private SessionState state;
	private long establishTimestamp;
	private long requestTimestamp;
	private AtomicInteger sequence;
	private boolean checkingHeartBeat = false;
	private Map<Object, Object> attributes;
	private Logger logger = LoggerFactory.getLogger(getClass());

	public MinaSessionImpl(MinaTcpTransporter parent, IoSession ioSession,
			SessionManager manager, boolean crypto) throws Exception {
		this.parent = parent;
		this.key = "MINA-" + ID.getAndIncrement();
		this.crypto = crypto;
		this.closed = false;
		this.ioSession = ioSession;
		this.sessionManager = manager;
		this.sequence = new AtomicInteger(0);
		this.attributes = new HashMap<Object, Object>();
		this.establishTimestamp = System.currentTimeMillis();
		this.requestTimestamp = -1;

		// 在session上绑定消息收发必须的属性对象
		prepareAttributes();
	}

	public MinaSessionImpl(MinaTcpTransporter parent, IoSession ioSession,
			SessionManager manager) throws Exception {
		this(parent, ioSession, manager, true);
	}

	private void prepareAttributes() throws Exception {
		// 设置消息接收器
		IoBufferMessageReceiver receiver = new IoBufferMessageReceiver(this);
		ioSession.setAttribute(SessionManager.ATTR_MESSAGE_RECEIVER, receiver);
	}

	@Override
	public String getKey() {
		return key;
	}

	@Override
	public boolean isCrypto() {
		return crypto;
	}

	@Override
	public boolean isClosed() {
		return closed;
	}

	@Override
	public long getEstablishTimestamp() {
		return this.establishTimestamp;
	}

	@Override
	public SessionManager getSessionManager() {
		return this.sessionManager;
	}

	@Override
	public SocketAddress getRemoteAddress() {
		return ioSession.getRemoteAddress();
	}

	@Override
	public void close() {
		if (closed) {
			return;
		}

		try {
			// 删除IoSession上附着的所有属性
			List<Object> keys = new ArrayList<Object>();
			keys.addAll(ioSession.getAttributeKeys());
			for (Object key : keys) {
				ioSession.removeAttribute(key);
			}

			// ！！！这里不能清除自己的attributes

			ioSession.close(true);
			closed = true;
		} catch (Exception e) {
			logger.error("Close session failed.", e);
		}
	}

	@Override
	public void setAttribute(Object key, Object value) {
		attributes.put(key, value);
	}

	@Override
	public Object getAttribute(Object key) {
		return attributes.get(key);
	}

	@Override
	public Object getAttribute(Object key, Object valueOnAbsent) {
		Object value = attributes.get(key);
		return value == null ? valueOnAbsent : value;
	}

	@Override
	public Object removeAttribute(Object key) {
		return attributes.remove(key);
	}

	@Override
	public SessionState getState() {
		return state;
	}

	@Override
	public void setState(SessionState state) {
		this.state = state;
	}

	@Override
	public long getRequestTimestamp() {
		return this.requestTimestamp;
	}

	@Override
	public void setRequestTimestamp(long timestamp) {
		this.requestTimestamp = timestamp;
	}

	@Override
	public int getReceiveSequence() {
		return sequence.get();
	}

	@Override
	public int incReceiveSequence() {
		return sequence.getAndIncrement();
	}

	@Override
	public void setCheckingHeartbeat(boolean check) {
		checkingHeartBeat = check;
	}

	@Override
	public boolean isCheckingHeartBeat() {
		return checkingHeartBeat;
	}

	@Override
	public void sendCommandResponse(CommandMessage cm) throws Exception {
		if (logger.isDebugEnabled()) {
			logger.debug(String.format(
					"Send command %d response output argument: %s",
					cm.getCommandId(), cm.getOutput().toString()));
		}

		sendMessage(cm);
	}

	@Override
	public void pushCommand(int commandId, MessageBean mbean) throws Exception {
		CommandMessage cm = (CommandMessage) Message.createMessage(
				Message.MT_APPLICATION, sessionManager.getCommandFactory());
		Command command = cm.createCommand(commandId);
		Class<?> outputClazz = command.getOutput().getClazz();
		if (outputClazz == null) {
			throw new TransportException(String.format(
					"command id=%d hasn't output argument.", commandId));
		} else if (mbean.getClass() != outputClazz) {
			throw new TransportException("Push command id " + commandId
					+ " message bean class " + mbean.getClass().getSimpleName()
					+ " isn't required " + outputClazz.getSimpleName());
		}

		if (logger.isDebugEnabled()) {
			logger.debug(String.format("Push command %d: %s", commandId,
					mbean.toString()));
		}
		command.getOutput().setMessageBean(mbean);
		sendMessage(cm);
	}

	@Override
	public void sendMessage(Message message) {
		try {
			// 分配IoBuffer，并设置成为可自动扩展。
			IoBuffer buffer = parent.getSendBuffer();

			// 预留一个int（4个字节）的MessageSize
			buffer.putInt(0);

			// 1个字节的MessageType
			buffer.put((byte) message.getType());

			// 把Message序列化到buffer中
			message.serialize(buffer.asOutputStream());

			// 回填MessageSize字段
			int msgsize = buffer.position() - 4;
			buffer.putInt(0, msgsize);

			// flip buffer
			buffer.flip();
//			
//			if (logger.isDebugEnabled() && (message instanceof CommandMessage)) {
//				CommandMessage cm = (CommandMessage) message;
//				logger.debug("SendMessage: " + cm.getCommandId() + " " + cm.getOutput().toString());
//				logger.debug("dataLength=" + buffer.limit());
//			}

			ioSession.write(buffer);
		} catch (Throwable t) {
			t.printStackTrace();
			throw new TransportException("send message error.", t);
		}
	}
}
